package org.uiot.process

import grails.async.Promise
import org.apache.flink.api.common.JobExecutionResult

class CodecJob implements StreamJobTraits, RmqInputTraits, RmqOutputTraits {

    Codec.Mode mode

    static constraints = {
        flinkInst()
        checkPointing()
        rmqInput()
        sourceQueue()
        rmqOutput()
        sinkQueue()
        mode()
    }

    @Override
    void submit(String operator) {
        final env = getStreamExecutionEnvironment()

        env.addSource(rmqSource).returns(String.class).setParallelism(1)
                .flatMap(Codec.getInstance(mode).cachePackets([]).cacheRules([]).cacheSequenceGroups([]))
                .returns(Map.class).writeUsingOutputFormat(rmqSink).name("CodecJob:$mode")

        final task = this

        grails.async.Promises.task {
            env.execute(this.name)
        }.then { Promise<JobExecutionResult> p ->
            //task.addToOperations(new Operation(operator: operator, job: ([jobId: p.get().jobID as String] as grails.converters.JSON) as String))
            task.save(flush: true)
        }
    }
}
